/**
 *
 */
package 线程.queue;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedTransferQueue;
import java.util.concurrent.TransferQueue;

import org.junit.Test;

/**
 * 当我们不想生产者过度生产消息时，TransferQueue可能非常有用，可避免发生OutOfMemory错误。
 * 在这样的设计中，消费者的消费能力将决定生产者产生消息的速度
 *
 * @Title:
 * @author zhoujie
 * @date 2018年7月25日 下午1:56:01
 *
 */
public class LinkedTransferQueueDemo {
    TransferQueue<String> transferQueue = new LinkedTransferQueue<>();
    ExecutorService ex = Executors.newCachedThreadPool();

    private void awart() {
        ex.shutdown();
        while (true) {
            if (ex.isTerminated()) {
                break;
            }
        }
    }

    @Test
    public void test01() {
        try {
            // transferQueue.transfer("123");//阻塞住 直到有消费者拿走
            transferQueue.put("123");// 有消费者就直接给 没有就写入队列
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        ex.submit(() -> {
            System.out.println(transferQueue.poll());
        });
        awart();
    }

    @Test
    public void test02() {
        int max = 100;

        for (int i = 0; i < max; i++) {
            ex.submit(() -> {
                try {
                    System.out.println(transferQueue.take());
                } catch (InterruptedException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }//阻塞获取
            });
        }

        for (int i = 0; i < max; i++) {
            // 有没被消耗的则阻塞，没有则加入
            try {
                if (transferQueue.size() > 0) {
                    transferQueue.transfer(i + "");
                } else {
                    transferQueue.put(i + "");
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            } // 有消费者就直接给 没有就写入队列
        }

        awart();
    }
}
